package com.shujia.flink.tf

import org.apache.flink.api.common.functions.FlatMapFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object Demo2FlatMap {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val linesDS: DataStream[String] = env.readTextFile("data/words.txt")

    //scala写法
    val ds1: DataStream[String] = linesDS.flatMap(line => line.split(","))

    //java的方式
    val ds2: DataStream[String] = linesDS.flatMap(new FlatMapFunction[String,String] {
      override def flatMap(t: String, collector: Collector[String]): Unit = {
        val split: Array[String] = t.split(",")

        for (word <- split) {

          //将数据输出到下游
          collector.collect(word)
        }
      }
    })

    ds2.print()

    env.execute()
  }
}
